package com.mo.redis;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

import java.util.Scanner;

/**
 * redis stream监听消息
 *
 * @author moon.pan
 */
@Component
@ConditionalOnProperty(value = "spring.redis.host")
public class ListenerMessage implements StreamListener<String, ObjectRecord<String, Object>> {

    @Autowired
    private StringRedisTemplate redisTemplate;

    private static final Scanner scanner = new Scanner(System.in);

    @Override
    public void onMessage(ObjectRecord<String, Object> message) {
        // 接收到消息
        System.out.println("message id " + message.getId());
        System.out.println("stream " + message.getStream());
        System.out.println("body " + new String((byte[]) message.getValue()));

        System.out.print("请输入选项 : ");
        if (scanner.hasNextBoolean()) {
            if (scanner.nextBoolean()) {
                redisTemplate.opsForStream().acknowledge("async_queue", "group1", message.getId());
                redisTemplate.opsForStream().delete("async_queue", message.getId());
                System.out.println("acked");
            }else {
                System.out.println("没有ack");
            }
        }
    }
}